package net.bwie.realtime.vehicle.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.common.utils.JdbcUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.vehicle.bean.VehicleData;
import net.bwie.realtime.vehicle.utils.DateTimeUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 异常检测与告警
 *
 * @Author: FuHe
 * @Date: 2025/5/27
 */
public class VehicleAbnormalAlarmJob {
    public static void main(String[] args) throws Exception {
        // 获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1
        env.setParallelism(1);
        // 从Kafka中消费数据
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle-date-geohash");
//        kafkaStream.print();
        // 处理数据，生成报警信息流
        DataStream<String> alarmStream = processheadel(kafkaStream);
        // 打印报警信息流
        alarmStream.print();
        // 将报警信息保存到ClickHouse数据库中
        JdbcUtil.sinkToClickhouseUpsert(alarmStream,
                "INSERT INTO vehicle_log.vehicle_abnormal_alarm (window_start, window_end, vin, alarm_type)\n" +
                        "VALUES (?, ?, ?, ?)");
        // 执行流处理任务
        env.execute("VehicleAbnormalAlarmJob");
    }

    /**
     * 处理输入流，生成车辆异常报警信息
     *
     * @param kafkaStream 输入的数据流，包含车辆数据的JSON字符串
     * @return 返回包含报警信息的数据流
     */
    private static DataStream<String> processheadel(DataStream<String> kafkaStream) {
        // 最低电池电量
        Double MinBatteryLevel = 10.0;
        // 最高电池温度
        Double MaxBatteryTemp = 60.0;
        // 最高电机温度
        Double maxMotorTemp = 60.0;
        // 最高速度
        Double maxSpeed = 120.0;
        // 解析JSON字符串为VehicleData对象
        SingleOutputStreamOperator<VehicleData> map = kafkaStream.map(o -> JSON.parseObject(o, VehicleData.class));
        // 分配时间戳和水印，用于窗口操作
        SingleOutputStreamOperator<VehicleData> vehicleDataSingleOutputStreamOperator = map.assignTimestampsAndWatermarks(WatermarkStrategy
                .<VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<VehicleData>() {
                            @Override
                            public long extractTimestamp(VehicleData element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }
                )
        );
        // 按照VehicleData对象进行分组
        KeyedStream<VehicleData, String> vehicleDataStringKeyedStream = vehicleDataSingleOutputStreamOperator.keyBy(o -> o.getVin());
        // 定义基于事件时间的滚动窗口，窗口大小为1分钟
        WindowedStream<VehicleData, String, TimeWindow> window = vehicleDataStringKeyedStream.window(
                TumblingEventTimeWindows.of(Time.minutes(1))
        );
        // 应用窗口函数，处理每个窗口的数据，生成报警信息
        SingleOutputStreamOperator<String> process = window.process(
                new ProcessWindowFunction<VehicleData, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<VehicleData> elements, Collector<String> out) throws Exception {
                        // 获取窗口的开始时间和结束时间
                        long window_start = context.window().getStart();
                        long window_end = context.window().getEnd();
                        // 初始化计数器
                        int currentCount = 0;

                        // 遍历窗口中的所有元素，统计当前异常情况发生的次数
                        for (VehicleData data : elements) {
                            // 电池电量报警
                            if (data.getBatteryLevel() != null && data.getBatteryLevel() < MinBatteryLevel) {
                                currentCount++;
                            }
                        }
                        // 如果异常次数大于等于3次，则输出报警
                        if (currentCount >= 3) {
                            out.collect(window_start + "," + window_end + "," + s + ", 电池电量低于10%");
                        }
                    }
                }
        );
        // 返回包含报警信息的数据流
        return process;
    }
}
